package cn._51doit.flink.day04;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

/**
 * 使用新的提取EventTime生成WaterMark的API
 */
public class EventTimeTumblingWindowDemoNewApi {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //新版本默认的时间类型即为EventTime
        //1000,spark,3
        //3000,hadoop,5

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //提取数据中的eventTime

        //assignTimestampsAndWatermarks该方法仅是提取数据中的EventTime，提取完返回的数据没有任何改变
        //处理完后的数据 ：1640844000000,spark,3
        SingleOutputStreamOperator<String> dataStreamWithWaterMark = lines.assignTimestampsAndWatermarks(
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String ele, long l) {
                        String[] fields = ele.split(",");
                        return Long.parseLong(fields[0]);
                    }
                }));

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataStreamWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String line) throws Exception {
                //line => 1640844000000,spark,3
                String[] fields = line.split(",");
                String word = fields[1];
                int count = Integer.parseInt(fields[2]);
                return Tuple2.of(word, count);
            }
        });

        //先KeyBy
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(tp -> tp.f0);

        //按照EventTime滚动的窗口，窗口长度为10秒
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyed.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> res = window.sum(1);

        res.print();

        env.execute();

    }
}
